// Copyright (c) 2012 SameGoal LLC.
// All Rights Reserved.
// Author: Andy Hochhaus <ahochhaus@samegoal.com>

// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "net/http/server/event/connection.h"

#include <arpa/inet.h>
#include <gflags/gflags.h>
#include <glog/logging.h>
#include <event2/buffer.h>
#include <event2/bufferevent.h>
#include <event2/bufferevent_ssl.h>
#include <event2/event.h>
#include <event2/listener.h>
#include <event2/util.h>
#include <openssl/err.h>
#include <openssl/rand.h>
#include <openssl/ssl.h>
#include <re2/re2.h>

#include <map>
#include <string>
#include <vector>
#include <functional>

#include "base/init.h"
#include "base/string.h"
#include "util/worker_pool/worker_pool.h"

namespace sg {
namespace net {
namespace http {
namespace server {
namespace event {

Connection::Connection(EventServer *server,
                       sockaddr_t *address,
                       bufferevent *buff_ev,
                       bool encrypted)
    : server_(server), buff_ev_(buff_ev), encrypted_(encrypted),
      read_timeout_(NULL), write_timeout_(NULL), closed_(false),
      close_cb_(nullptr), connection_mode_(UNKNOWN) {
  CHECK_NOTNULL(server_);
  CHECK_NOTNULL(address);
  CHECK_NOTNULL(buff_ev_);
  CHECK_EQ(AF_INET, address->sa_family);
  sockaddr_in *address_in = reinterpret_cast<sockaddr_in *> (address);
  char buf[INET_ADDRSTRLEN];
  if (!inet_ntop(address->sa_family,
                 &address_in->sin_addr,
                 buf,
                 INET_ADDRSTRLEN)) {
    address_ = "unknown";
  } else {
    address_ = std::string(buf);
  }
  port_ = ntohs(address_in->sin_port);
  requests_.push_back(new Request(this));
}

Connection::~Connection() {
  delete read_timeout_;
  delete write_timeout_;
  while (requests_.size() > 0) {
    Request *request = requests_.front();
    requests_.pop_front();
    delete request;
  }
  while (requests_to_delete_.size() > 0) {
    Request *request = requests_to_delete_.front();
    requests_to_delete_.pop_front();
    delete request;
  }
}

bool Connection::encrypted() const {
  return encrypted_;
}

void Connection::Init() {
  bool should_delete = false;
  {
    std::lock_guard<std::recursive_mutex> lock(mutex_);
    bufferevent_setcb(
        buff_ev_, &ReadCallback, &WriteCallback, &EventCallback, this);
    if (bufferevent_set_timeouts(
            buff_ev_, read_timeout_, write_timeout_) != 0) {
      LOG(ERROR) << "unable to set timeouts.";
      should_delete = CloseConnection();
      goto error;
    }
    if (bufferevent_enable(buff_ev_, EV_READ | EV_WRITE | EV_PERSIST) != 0) {
      LOG(ERROR) << "unable to bufferevent_enable";
      should_delete = CloseConnection();
    }
  }

 error:
  if (should_delete) {
    LOG(INFO) << "delete connection: " << this;
    delete this;
  }
}

std::string Connection::address() {
  return address_;
}

uint16 Connection::port() {
  return port_;
}

void Connection::set_read_timeout(const timeval_t &timeout) {
  delete read_timeout_;
  read_timeout_ = NULL;
  read_timeout_ = new timeval_t(timeout);
  CHECK_NOTNULL(read_timeout_);
}

void Connection::set_write_timeout(const timeval_t &timeout) {
  delete write_timeout_;
  write_timeout_ = NULL;
  write_timeout_ = new timeval_t(timeout);
  CHECK_NOTNULL(write_timeout_);
}

EventServer *Connection::event_server() {
  return server_;
}

std::recursive_mutex *Connection::mutex() {
  return &mutex_;
}

void Connection::OutputReadyEvent(Request */* req */) {
  bool should_delete = false;
  {
    std::lock_guard<std::recursive_mutex> lock(mutex_);
    if (!closed_) {
      if (event_base_once(
              bufferevent_get_base(buff_ev_),
              -1, EV_TIMEOUT, &Connection::WriteOutput, this, NULL) != 0) {
        LOG(ERROR) << "unable to schedule WriteOutput";
        should_delete = CloseConnection();
      }
    } else {
      while (!requests_.empty() && requests_.front()->IsFinished()) {
        requests_to_delete_.push_back(requests_.front());
        requests_.pop_front();
      }
      if (requests_.empty()) {
        should_delete = true;
      }
    }
  }

  if (should_delete) {
    LOG(INFO) << "delete connection: " << this;
    delete this;
  }
}

void Connection::SetConnectionCloseHandler(Callback callback) {
  close_cb_ = callback;
}

void Connection::CloseConnectionAfterNextWrite() {
  std::lock_guard<std::recursive_mutex> lock(mutex_);
  connection_mode_ = CLOSE;
}

void Connection::ReadCallback(bufferevent_t */* buff_ev */, void *ctx) {
  bool should_delete = false;
  Connection *connection = static_cast<Connection *> (ctx);
  {
    std::lock_guard<std::recursive_mutex> lock(connection->mutex_);
    evbuffer_t *input = bufferevent_get_input(connection->buff_ev_);

    bool created_new_request;
    do {
      created_new_request = false;
      Request *req = connection->requests_.back();
      if (req->IsReadingHeaders()) {
        if (!connection->ReadHeaders(req, input)) {
          LOG(ERROR) << "unable to ReadHeaders";
          should_delete = connection->CloseConnection();
          goto error;
        }
      }
      if (req->IsReadingBody()) {
        connection->ReadBody(req, input);
      }
      if (req->IsReadingDone()) {
        ConnectionMode mode = GetConnectionType(req);
        if (connection->connection_mode_ == UNKNOWN) {
          connection->connection_mode_ = mode;
        }
        if (connection->connection_mode_ != mode) {
          LOG(ERROR) << "bad connection type: " << mode
                     << " over type: " << connection->connection_mode_;
        }
        connection->server_->app_callback_pool.Add(
            std::bind(&EventServer::DispatchRequest,
                      connection->server_,
                      std::placeholders::_1),
            req);
        if (connection->connection_mode_ != SINGLE) {
          connection->requests_.push_back(new Request(connection));
          created_new_request = true;
        }
      }
    } while (created_new_request);
  }

 error:
  if (should_delete) {
    LOG(INFO) << "delete connection: " << connection;
    delete connection;
  }
}

void Connection::WriteCallback(bufferevent_t */* buff_ev */, void *ctx) {
  bool should_delete = false;
  Connection *connection = static_cast<Connection *> (ctx);
  {
    std::lock_guard<std::recursive_mutex> lock(connection->mutex_);
    if (connection->connection_mode_ == CLOSE ||
        (connection->requests_.empty() &&
         connection->connection_mode_ == SINGLE)) {
      should_delete = connection->CloseConnection();
    }
  }

  if (should_delete) {
    LOG(INFO) << "delete connection: " << connection;
    delete connection;
  }
}

void Connection::EventCallback(bufferevent_t */* buff_ev */,
                               int16  events,
                               void *ctx) {
  bool should_delete = false;
  Connection *connection = static_cast<Connection *> (ctx);
  {
    std::lock_guard<std::recursive_mutex> lock(connection->mutex_);

    if (events & BEV_EVENT_TIMEOUT) {
      VLOG(1) << "[" << connection << "] event timeout\n";
    }
    if (events & BEV_EVENT_CONNECTED) {
      VLOG(1) << "[" << connection << "] event connected\n";
    }
    if (events & BEV_EVENT_ERROR) {
      VLOG(1) << "[" << connection << "] event error\n";
      const int socket_errno = EVUTIL_SOCKET_ERROR();
      if (socket_errno) {
        LOG(ERROR) << "socket error: "
                   << evutil_socket_error_to_string(socket_errno);
      }
      auto err = bufferevent_get_openssl_error(connection->buff_ev_);
      if (err != 0) {
        LOG(ERROR) << "openssl error: "
                   << ERR_reason_error_string(err)
                   << " " << ERR_lib_error_string(err)
                   << " " << ERR_func_error_string(err);
      }
    }
    if (events & BEV_EVENT_EOF) {
      VLOG(1) << "[" << connection << "] event eof\n";
    }

    if (events & (BEV_EVENT_ERROR | BEV_EVENT_EOF | BEV_EVENT_TIMEOUT)) {
      should_delete = connection->CloseConnection();
    }
  }

  if (should_delete) {
    LOG(INFO) << "delete connection: " << connection;
    delete connection;
  }
}

void Connection::WriteOutput(
    evutil_socket_t /* fd */, int16 /* events */, void *ctx) {
  bool should_delete = false;
  Connection *connection = static_cast<Connection *> (ctx);
  {
    std::lock_guard<std::recursive_mutex> lock(connection->mutex_);
    switch (connection->connection_mode_) {
      case UNKNOWN:
        // In the event of an HTTP/1.1 Expect, the current request may be
        // blocking on the client for the server to send a 100 Continue
        // response and the connection has an unknown mode.
      case SINGLE:
      case ORDERED:
      case CLOSE:
        while (!connection->requests_.empty()) {
          std::string output;
          Request *request = connection->requests_.front();
          request->GetOutput(&output);
          if (!output.empty() && !connection->closed_) {
            if (bufferevent_write(connection->buff_ev_,
                                  output.c_str(),
                                  output.length()) != 0) {
              LOG(ERROR) << "unable to write output to buff_ev_";
              should_delete = connection->CloseConnection();
              goto error;
            }
          }
          if (request->IsFinished()) {
            connection->requests_to_delete_.push_back(
                connection->requests_.front());
            connection->requests_.pop_front();
          } else {
            break;
          }
        }
        break;
      default:
        LOG(ERROR) << "unsupported connection_mode_: "
                   << connection->connection_mode_;
        should_delete = connection->CloseConnection();
    }
  }

 error:
  if (should_delete) {
    LOG(INFO) << "delete connection: " << connection;
    delete connection;
  }
}

bool Connection::ReadHeaders(Request *req, evbuffer_t *input) {
  size_t len;
  char *line;
  while (req->IsReadingHeaders() &&
         (line = evbuffer_readln(input, &len, EVBUFFER_EOL_CRLF)) != NULL) {
    if (!req->AddHeader(std::string(line, len))) {
      free(line);
      return false;
    }
    free(line);
  }
  return true;
}

bool Connection::ReadBody(Request *req, evbuffer_t *input) {
  std::string transfer_encoding;
  req->GetHeader("Transfer-Encoding", &transfer_encoding);
  uint64 content_length;
  bool has_content_length = req->GetHeaderAsUint64("Content-Length",
                                                   &content_length);
  if (transfer_encoding == "chunked") {
    // TODO(ahochhaus): Maybe fire the application level callback early in the
    // case of a chunked request body and fire events as additional data is
    // uploaded. The current implementation buffers the request body until all
    // of the data is uploaded and then fires the application level callback.
    bool read_chunk;
    do {
      read_chunk = false;
      size_t eol_len_out;
      evbuffer_ptr_t eol_ptr = evbuffer_search_eol(
          input, NULL, &eol_len_out, EVBUFFER_EOL_CRLF);
      if (eol_ptr.pos < 0) {
        break;
      }
      unsigned char *len_prefix_line = evbuffer_pullup(input, eol_ptr.pos);
      if (len_prefix_line == NULL) {
        LOG(ERROR) << "unable to read len_prefix_line of size: "
                   << eol_ptr.pos;
        return false;
      }
      std::string len_prefix(reinterpret_cast<char *> (len_prefix_line),
                             eol_ptr.pos);
      uint64 len;
      if (!sg::string::FromHexString(len_prefix, &len)) {
        LOG(ERROR) << "invalid chunk len prefix: " << len_prefix;
        return false;
      }
      const uint64 total_size = eol_ptr.pos + eol_len_out + len;
      const size_t bytes_available = evbuffer_get_length(input);
      if (bytes_available >= total_size) {
        std::string body;
        body.resize(content_length);
        // c++11 strings are guaranteed to be contiguous
        if (evbuffer_remove(input, &body[0], content_length) !=
            static_cast<int64> (content_length)) {
          LOG(ERROR) << "error reading request body of length: "
                     << content_length;
          return false;
        }
        if (len == 0) {
          req->AppendBody("", true);
        } else {
          req->AppendBody(body, false);
        }
      }
    } while (read_chunk);
  } else if (has_content_length && content_length > 0) {
    const size_t bytes_available = evbuffer_get_length(input);
    if (bytes_available >= content_length) {
      std::string body;
      body.resize(content_length);
      // c++11 strings are guaranteed to be contiguous
      if (evbuffer_remove(input, &body[0], content_length) !=
          static_cast<int64> (content_length)) {
        LOG(ERROR) << "error reading request body of length: "
                   << content_length;
        return false;
      }
      req->AppendBody(body, true);
    }
  } else {
    // no request body
    req->AppendBody("", true);
  }
  return true;
}

bool Connection::CloseConnection() {
  if (!closed_) {
    if (close_cb_) {
      for (auto it = requests_.begin(); it != requests_.end(); ++it) {
        close_cb_(*it, NULL);
      }
    }
    closed_ = true;
    bufferevent_free(buff_ev_);
    buff_ev_ = NULL;
  }
  // Terminate partial incoming requests. The underlying bufferevent is now
  // closed so the requests will never fully arrive.
  while (!requests_.empty() && !requests_.back()->IsReadingDone()) {
    requests_to_delete_.push_back(requests_.back());
    requests_.pop_back();
  }
  if (requests_.empty()) {
    return true;
  }
  return false;
}

// HTTP requests interact with the underlying TCP connection in different ways.
//
// 1 HTTP request per TCP connection
//   * HTTP/1.0 without a Connection header
//   * HTTP/1.1 with a "Connection: close" header
//   - connection_mode_ = SINGLE
//
// HTTP Persistent connection: back-to-back HTTP requests over TCP connection
//                             (only one request is active at a time)
//   * HTTP/1.0 with a "Connection: Keep-Alive" header
//   * HTTP/1.1 without a Connection header [See note in pipelining section]
//   - connection_mode_ = ORDERED (same as pipelined)
//
// HTTP Pipelining: concurrent HTTP requests per TCP connection
//                  (multiple requests are in-flight concurrently -- however
//                  the requests and responses are ordered)
//                  http://www.blaze.io/mobile/http-pipelining-big-in-mobile/
//   * HTTP/1.1 with a response "Connection: Keep-Alive" header [Opera/Android]
//   - connection_mode_ = ORDERED (same as persistent)
//
// HTTP Multiplexing: concurrent unordered HTTP requests per TCP connection
//                    (multiple requests are in-flight concurrently with no
//                    ordering constraints)
//   * SPDY (SSL handshake Next Protocol Negotiation; NPN) [Chrome/Firefox]
//   - connection_mode_ = MULTIPLEXED
Connection::ConnectionMode Connection::GetConnectionType(Request *req) {
  std::string connection;
  req->GetHeader("Connection", &connection);
  const Request::RequestVersion request_version = req->request_version();

  // TODO(ahochhaus): support MULTIPLEXED

  if (connection == "Keep-Alive" ||
      (connection != "Close" && request_version == Request::ONE_ONE)) {
    return ORDERED;
  }

  // fall back to the safest protocol
  return SINGLE;
}

}  // namespace event
}  // namespace server
}  // namespace http
}  // namespace net
}  // namespace sg
